package com.flink.demo;


import java.io.IOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.sql.Timestamp;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 数据生成器 - 模拟用户行为数据
 * 运行方式: java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.DataGenerator
 */
public class DataGenerator {
    private static final String[] BEHAVIORS = {"click", "buy", "cart", "pv"};
    private static final String[] CATEGORIES = {"electronics", "clothing", "food", "books", "sports"};
    private static final Random random = new Random();

    public static void main(String[] args) throws IOException {
        int port = 9999;
        ServerSocket serverSocket = new ServerSocket(port);
        System.out.println("数据生成器启动在端口: " + port);

        while (true) {
            try (Socket clientSocket = serverSocket.accept();
                 OutputStream out = clientSocket.getOutputStream()) {

                System.out.println("客户端连接成功，开始生成数据...");

                while (!clientSocket.isClosed()) {
                    // 生成模拟数据
                    UserBehavior behavior = generateUserBehavior();
                    String data = behavior + "\n";

                    // 发送数据
                    out.write(data.getBytes());
                    out.flush();

                    System.out.println("发送数据: " + data.trim());

                    // 随机延迟 0.5-2秒
                    TimeUnit.MILLISECONDS.sleep(500 + random.nextInt(1500));
                }
            } catch (Exception e) {
                System.out.println("客户端断开连接，等待新连接...");
            }
        }
    }

    private static UserBehavior generateUserBehavior() {
        Long userId = (long) (1 + random.nextInt(100));  // 1-100的用户
        String itemId = "item_" + (1 + random.nextInt(1000));
        String category = CATEGORIES[random.nextInt(CATEGORIES.length)];
        String behavior = BEHAVIORS[random.nextInt(BEHAVIORS.length)];
        Timestamp timestamp = new Timestamp(System.currentTimeMillis());

        return new UserBehavior(userId, itemId, category, behavior, timestamp);
    }
}